package edu.gmu.server;

import edu.gmu.messages.EchoRequest;
import edu.gmu.messages.EchoResponse;
import edu.gmu.middleware.Header;
import edu.gmu.middleware.Marshaler;
import edu.gmu.middleware.MarshalerImpl;
import edu.gmu.middleware.Message;
import edu.gmu.middleware.MessageBinding;
import edu.gmu.middleware.MessageRegistry;
import edu.gmu.middleware.Sink;
import edu.gmu.middleware.SinkDataEvent;
import edu.gmu.middleware.SinkDisconnectedEvent;
import edu.gmu.middleware.SinkFactory;
import edu.gmu.middleware.SinkListener;
import edu.gmu.middleware.Unmarshaler;
import edu.gmu.middleware.UnmarshalerImpl;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

public class TCPEchoServer
{
    private ServerSocket serverSocket = null;
    private int port = 0;
    private volatile boolean stopped = false;
    
    private List clients = new ArrayList();
    
    //private ByteBuffer in  = ByteBuffer.allocate(1048576);
    private ByteBuffer out = ByteBuffer.allocate(1048576);
    
    private Marshaler marshaler      = new MarshalerImpl();
    private Unmarshaler unmarshaler  = new UnmarshalerImpl();
    
    private MessageRegistry registry = new MessageRegistry();
    
    /**
     * Creates a new TCPEchoServer with specified params.
     */
    public TCPEchoServer( )
    {
        this.registry.register(100, EchoRequest.class);
    }
    
    /**
     * Starts the server listening on specified port - blocks caller's thread
     * until finished.
     * @param int port
     * @throws java.io.IOException
     */
    public void listen( int port ) throws IOException
    {
        this.port = port;
        this.serverSocket = new ServerSocket(this.port);
        
        this.stopped = false;
        while( stopped == false )
        {
            // Blocks until client connects
            Socket clientSocket = this.serverSocket.accept();
            System.out.println("TCPEchoServer - Client connected.");
            this.handleClient(clientSocket);
        }
    }
    
    /**
     * Handles a new client connection.
     * @param Socket clientSocket
     */
    public void handleClient( Socket clientSocket )
    {
        //--------------------------------------------------------------------//
        // Be quick here so that server can go back to listening
        //--------------------------------------------------------------------//
        try
        {
            final Sink sink = SinkFactory.makeTCPSink(clientSocket,1048576 ); //1048576  65536
            
            sink.addSinkListener(new SinkListener()
                {
                    public void dataReceived( SinkDataEvent event )
                    {
                        try
                        {
                            ByteBuffer in = event.getIn();
                            Header header = event.getHeader();
                            //print(in,12);
                            int topic = header.getTopic();
                            MessageBinding messageBinding = registry.getMessageBinding(topic);
                            if( messageBinding != null )
                            {
                                Message message = unmarshaler.unmarshal(header,
                                                                    in,
                                                                    messageBinding);
                        
                                EchoRequest request = (EchoRequest)message;

                                EchoResponse response = new EchoResponse();
                                response.count   = request.count++;
                                response.missive = request.missive;

                                // Just use same header to marshal
                                header.setTopic(101);
                                marshaler.marshal(header, response, out);
                                sink.write(out);
                            }
                            else
                            {
                                System.out.println( "No messageBinding for topic "+topic );
                            }
                        }
                        catch (IOException ex)
                        {
                            ex.printStackTrace();
                        }
                    }
                    
                    public void sinkDisconnected( SinkDisconnectedEvent event )
                    {
                        // Regardless of disconnect type - go ahead and shutdown
                        // We know which client - therefore remove and close

                        // To be quite honest - not sure what "this" means here?
                        sink.removeSinkListener(this);

                        System.out.println("Handling event in TCPEchoServer.");
                        event.getException().printStackTrace();

                        clients.remove(sink);
                        sink.disconnect();
                        System.out.println("   clients.size()="+clients.size());
                    }
                }
            
                
            );
            
            this.clients.add(sink);
            sink.connect();
        }
        catch(IOException ioe)
        {
            ioe.printStackTrace();
        }
        //--------------------------------------------------------------------//
    }
    
    /**
     * Stops the server from listening
     * @throws java.io.IOException
     */
    public void stop()
    {
        this.stopped = true;
        // Stop server - so that we accept no more clients
        if( this.serverSocket != null )
        {
            try
            {
                this.serverSocket.close();
            }
            catch(IOException ioe)
            {
                // ignore
            }
        }
        
        // Stop clients
        for (int i = 0; i < clients.size(); i++)
        {
            Sink sinkManager = (Sink)clients.get(i);
            sinkManager.disconnect();
        }
    }
 
    
    /**
     * Solely to be used for debugging.
     * @param inBuffer
     * @param length
     */
    private static void print( ByteBuffer inBuffer, int length )
    {
        byte[] in = inBuffer.array();
        System.out.print( "position="+inBuffer.position() + " limit="+inBuffer.limit() );
        for (int j = 0; j < length; j++)
        {
            String s = Integer.toHexString( in[j]&0xff );
            if( s.length() == 1 ) s = "0"+s;
            System.out.print( " " + s );
        }
        System.out.println( "" );
    }
    
    
    public static void main( String[] args ) throws IOException
    {
        int localPort  = 5000;
        
        TCPEchoServer server = new TCPEchoServer();
        server.listen(localPort);
    }
    

}
